package com.bfxy.kafka.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Service
public class KafkaProducerService {

	private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducerService.class);

	@Resource
	private KafkaTemplate<String, Object> kafkaTemplate;
	
	public void sendMessage(String topic, Object object) {
		ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, object);
		future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
			@Override
			public void onSuccess(SendResult<String, Object> result) {
				LOGGER.info("发送消息成功: " + result.toString());
			}

			@Override
			public void onFailure(Throwable throwable) {
				LOGGER.error("发送消息失败: " + throwable.getMessage());
			}
		});
	}
	
}
